// Copyright 2010, Vanya Davidenko.
// Используемая кодировка: UTF-8.

#define _SCL_SECURE_NO_WARNINGS

// Данный хедер включает также поддержку TryCriticalSection.
// Он должен идти самым первым, чтобы windows.h была подключена с заданными
// настройками (через макросы), а не проигнорирована (из-за header guard).
# ifdef _WIN32
#  include <libv/windows_lean.h>
# endif  // _WIN32

#ifdef TESTING
# include "common.h"
# include <gtest/gtest.h>
# include <testing/base64_testing.h>
# include <testing/curl_testing.h>
# include <testing/html_text_extractor_testing.h>
# include <testing/url_converter_testing.h>
# include <testing/set_recognizer_testing.h>
# include <testing/map_replacer_testing.h>

#else

# ifdef _WIN32
#  include <direct.h>
# else
#  include <sys/stat.h>
#  include <sys/types.h>
# endif

# include <errno.h>
# include <omp.h>

# include <algorithm>
# include <list>
# include <fstream>
# include <sstream>

# include <tclap/CmdLine.h>
# include <tclap/StdOutput.h>

# include <libv/atomic.h>
# include <libv/check.h>
# include <libv/error_msg.h>
# include <libv/rdtsc.h>
# include <libv/sleep.h>
# include <libv/uncopyable.h>

# include "base64.h"
# include "common.h"
# include "icu_init.h"
# include "manager.h"
# include "socket_sender.h"
# include "url_converter.h"

#endif  // TESTING

namespace scrw
{

/** Уровень подробности отладочных сообщений (0 — выводить только ошибки). */
int g_verbose_level = 0;

/** Блокировка вывода отладочных сообщений в стандартные потоки вывода. */
libv::Spin g_console_lock;

}


#ifdef TESTING

namespace scrw
{

int Main(int argc, char **argv)
{
//     argc = 2;
//     char* argv_init[2] = { NULL, "--gtest_filter=*UrlCo*" };
//     argv = argv_init;

    g_verbose_level = -1;  // вообще не выводить ошибки
                           // (которые могут намеренно заноситься тестами).
    ::testing::InitGoogleTest(&argc, argv);
    int result = RUN_ALL_TESTS();
    return result;
}

}  // scrw

#else

namespace scrw
{

class Program : libv::Uncopyable
{
  public:
    Program()
        : manager_(NULL),
          save_to_socket_(false),
          max_reconnect_attempts_(1),
          download_pause_(1),
          save_to_folder_(false),
          file_base_(libv::rdtsc()),
          file_index_(0),
          continue_jobs_(1)
    {
        // В многопоточной среде явный вызов обязателен.
        TheICUInitializationAndCleanUp::Instance();

        utf8_bom_.push_back(static_cast<uint8_t>(0xEF));
        utf8_bom_.push_back(static_cast<uint8_t>(0xBB));
        utf8_bom_.push_back(static_cast<uint8_t>(0xBF));

        mime2ext_["application/pdf"] = "pdf";
        mime2ext_["application/msword"] = "doc";
        mime2ext_["application/vnd.openxmlformats-officedocument."
                  "wordprocessingml.document"] = "docx";
        mime2ext_["application/vnd.ms-excel"] = "xls";
        mime2ext_["application/vnd.openxmlformats-officedocument."
                  "spreadsheetml.sheet"] = "xlsx";
        mime2ext_["application/vnd.ms-powerpoint"] = "ppt";
        mime2ext_["application/vnd.openxmlformats-officedocument."
                  "presentationml.presentation"] = "pptx";
        mime2ext_["application/rtf"] = "rtf";

        mime2ext_["text/html"] = "htm";
        mime2ext_["text/plain"] = "txt";

        mime2ext_["text/xml"] = "xml";
        mime2ext_["application/xml"] = "xml";
        mime2ext_["application/atom+xml"] = "xml";
        mime2ext_["application/rdf+xml"] = "xml";
        mime2ext_["application/rss+xml"] = "xml";
        mime2ext_["application/xhtml+xml"] = "xml";
    }


    ~Program()
    {
        if ( save_type_ == SAVE_URLS_ONLY && urls_file_.is_open() )
        {
            urls_file_.exceptions(0);
            urls_file_.flush();  // игнорируем ошибки.
            urls_file_.close();
        }

        for ( ::std::vector<Manager::Extractor*>::iterator \
              i = extractors_.begin() ; i != extractors_.end() ; ++i )
        {
            delete *i;
            *i = NULL;
        }

        for ( ::std::vector<Manager::Downloader*>::iterator \
              i = downloaders_.begin() ; i != downloaders_.end() ; ++i )
        {
            delete *i;
            *i = NULL;
        }

        delete manager_;
        manager_ = NULL;
    }


    int ParseArguments(int argc, char** argv)
    {
        try
        {
            TCLAP::CmdLine cmd(
                "Simple crawler by Vanya Davidenko for iu9-bmstu",
                ' ', "1.5.4");

            SortedTclapOutput o;
            cmd.setOutput(&o);

            TCLAP::UnlabeledValueArg< ::std::string > arg(
                "url",
                "Url with max depth separated with space or "
                "filename of utf-8 file with such lines",
                true, "", "url max_depth  |  filename", cmd);

            TCLAP::SwitchArg save_as_plain_text(
                "", "save-as-plain-text",
                "Save pages as plain text with urls at begin in UTF-8.");

            TCLAP::SwitchArg save_urls_only(
                "", "save-urls-only",
                "Extract and save only urls from crawled pages.");

            TCLAP::SwitchArg save_as_xml(
                "", "save-as-xml",
                "Save pages as XML in UTF-8.");

            TCLAP::SwitchArg save_original_files(
                "", "save-original-files",
                "Save original files.");

            TCLAP::ValueArg< ::std::string > proxy(
                "p", "proxy",
                "Proxy settings. (default: none).",
                false, ::std::string(), "host[:port]", cmd);

            TCLAP::MultiSwitchArg verbose("v", "verbose",
                "Be more verbose.", cmd, 0);

            TCLAP::ValueArg< ::std::string > include_matcher(
                "i", "include-pattern",
                "Include matcher regex. "
                "Pages with such urls will be processed. (default: ^.*$).",
                false, "^.*$", "regex", cmd);

            TCLAP::ValueArg< ::std::string > exclude_matcher(
                "x", "exclude-pattern",
                "Exclude matcher regex. "
                "Pages with such address will be excluded from processing. "
                "(default ^$).",
                false, "^$", "regex", cmd);

            TCLAP::ValueArg< ::std::string > yield_matcher(
                "y", "yield-pattern",
                "Yield pages matcher. Pages with such urls will be outputed."
                "(default ^.*$).",
                false, "^.*$", "regex", cmd);

            TCLAP::ValueArg<size_t> hash_table_size(
                "", "urls-table-size",
                "Size of hash table that stores known urls,"
                "visited and enqueued to visit (default: 524287).",
                false, size_t(524287), "value", cmd);

            TCLAP::SwitchArg grab_nofollow(
                "", "grab-nofollow",
                "Grab also 'nofollow' links.", cmd, false);

            TCLAP::ValueArg< ::std::string > user_agent(
                "", "user-agent",
                "User agent (default: scrw (based on libcurl)).",
                false, "scrw (based on libcurl)", "string", cmd);

            TCLAP::ValueArg< ::std::string > referer(
                "", "referer",
                "Referer.", false, "", "string", cmd);

            TCLAP::ValueArg< ::std::string > cookies(
                "", "cookies",
                "Cookies to send, format: \"name1=var1;name2=var2\".",
                false, "", "string", cmd);

            TCLAP::ValueArg<size_t> connect_timeout(
                "", "connect-timeout",
                "Connect timeout (default: 30).", false, 30, "seconds", cmd);

            TCLAP::ValueArg<size_t> max_page_download_time(
                "", "max-page-download-time",
                "Max page download time (default: 30).",
                false, 30, "seconds", cmd);

            TCLAP::ValueArg<size_t> min_page_download_speed(
                "", "min-page-download-speed",
                "Min page download speed during 10 seconds (default: 5000).",
                false, 5000, "b/s", cmd);


            TCLAP::ValueArg<size_t> num_downloaders(
                "d", "num-downloaders",
                "Num of downloader threads (default: 1).",
                false, 1, "value", cmd);

            TCLAP::ValueArg<uint32_t> download_pause(
                "", "pause",
                "Pause between 2 downloads in one thread (default: 1).",
                false, 1, "seconds", cmd);

            TCLAP::ValueArg<size_t> max_num_redirects(
                "", "max-num-redirects",
                "Max number of redirects per page (default: 10).",
                false, 10, "num", cmd);

            TCLAP::ValueArg<size_t> num_extractors(
                "e", "num-extractors",
                "Num of extractors (default: 1).",
                false, 1, "value", cmd);


            TCLAP::ValueArg< ::std::string > folder(
                "f", "folder",
                "Folder to save files to.", false, "", "absolute path", cmd);

            TCLAP::ValueArg< ::std::string > socket(
                "s", "socket",
                "Socket to send files to.", false, "", "server:port", cmd);

            TCLAP::ValueArg<long> socket_timeout(
                "", "socket-timeout",
                "Socket connection timeout (default: 60).",
                false, 60, "seconds", cmd);

            TCLAP::ValueArg<long> socket_reconnect_attempts(
                "", "socket-reconnect-attempts",
                "Max socket reconnect attempts during file sending "
                "(default: 3).",
                false, 3, "num", cmd);

            ::std::vector<TCLAP::Arg*> xor_args;
            xor_args.push_back(&save_as_plain_text);
            xor_args.push_back(&save_as_xml);
            xor_args.push_back(&save_urls_only);
            xor_args.push_back(&save_original_files);
            cmd.xorAdd(xor_args);

            cmd.parse(argc, argv);

            if ( folder.isSet() == false && socket.isSet() == false )
            {
                ERR("You must specify at least one type of output: "
                    "folder or socket.")
                return -1;
            }


            bool extract_text = true;
            if ( save_as_xml.isSet() ) 
            {
                save_type_ = SAVE_AS_XML;
            }
            else if ( save_as_plain_text.isSet() )
            {
                save_type_ = SAVE_AS_PLAIN_TEXT;
            }
            else if ( save_urls_only.isSet() )
            {
                save_type_ = SAVE_URLS_ONLY;
                extract_text = false;
            }
            else if ( save_original_files.isSet() )
            {
                save_type_ = SAVE_ORIGINAL_FILES;
                extract_text = false;
            }
            else
            {
                assert(0);
            }

            
            // Должна идти самой первой, 
            // чтобы выводились ошибки создания объектов.
            g_verbose_level = verbose.getValue();

            if ( CreateManager(arg.getValue(),
                               proxy.getValue(),
                               include_matcher.getValue(),
                               exclude_matcher.getValue(),
                               yield_matcher.getValue(),
                               hash_table_size.getValue(),
                               grab_nofollow.getValue(),
                               extract_text,
                               user_agent.getValue(),
                               referer.getValue(),
                               cookies.getValue(),
                               connect_timeout.getValue(),
                               max_page_download_time.getValue(),
                               min_page_download_speed.getValue(),
                               max_num_redirects.getValue()) == -1 )
                return -1;

            if ( CreateDownloaders(num_downloaders.getValue(),
                                   (uint32_t)download_pause.getValue()) )
                return -1;

            if ( CreateExtractors(num_extractors.getValue()) == -1 )
                return -1;

            if ( folder.isSet() && CreateFolder(folder.getValue()) == -1 )
                return -1;

            if ( socket.isSet() &&
                 ConnectToSocket(socket.getValue(),
                                 socket_timeout.getValue(),
                                 socket_reconnect_attempts.getValue()) == -1 )
                return -1;

            if ( save_type_ == SAVE_URLS_ONLY && 
                 save_to_folder_ && CreateUrlsFile() == -1 )
                return -1;

            return 0;
        }
        catch(TCLAP::ArgException &e)
        {
            ::std::cerr << "Error: " << e.error() 
                        << " for arg " << e.argId() << ::std::endl;
            return -1;
        }
        catch(::std::exception& e)
        {
            ::std::cerr << "ParseArguments() caught unexpected exception: " 
                        << e.what() << ::std::endl;
            return -1;
        }
    }


    int Do()
    {
        const int num_threads = 1 +
                                libv::IntegerCast<int>(downloaders_.size()) +
                                libv::IntegerCast<int>(extractors_.size());

        int return_code = 0;
        #pragma omp parallel num_threads(num_threads)
        {
            if ( IsManager() )
            {
                return_code = ManagerCycle();
            }

            else if ( IsDownloader() )
            {
                DownloaderCycle();
            }

            else  // Extractor
            {
                ExtractorCycle();
            }
        }

        return return_code;
    }



  private:
    enum { SAVE_AS_XML, SAVE_AS_PLAIN_TEXT,
           SAVE_URLS_ONLY, SAVE_ORIGINAL_FILES };

    /** TCLAP не сортирует usage по ключам.
      * Данная реализация исправляет это. */
    class SortedTclapOutput : public TCLAP::StdOutput
    {
	  public:
        virtual void usage(TCLAP::CmdLineInterface& c)
        {
            std::cout << "\nUSAGE: \n" << std::endl;
            this->ShortUsage(c, std::cout);
            std::cout << "\n\nWhere: \n" << std::endl;
            this->LongUsage(c, std::cout);
            std::cout << std::endl;
        }

      private:
        class ArgFlagLess
        {
          public:
            typedef const TCLAP::Arg* first_argument_type;
            typedef const TCLAP::Arg* second_argument_type;
            typedef bool result_type;

            /** Сортировка аргументов:
              *   - обязательные идут перед опциональными
              *   - флаги идут перед длинными параметрами
              *   - все параметры отсортированы лексикографически. */
            bool operator()(const TCLAP::Arg* lhv, const TCLAP::Arg* rhv) const
            {
                if ( lhv->isRequired() && !rhv->isRequired() )
                    return true;
                else if ( !lhv->isRequired() && rhv->isRequired() )
                    return false;

                const bool lhv_is_a_flag = !lhv->getFlag().empty();
                const bool rhv_is_a_flag = !rhv->getFlag().empty();

                if ( lhv_is_a_flag && rhv_is_a_flag ) 
                    return lhv->getFlag() < rhv->getFlag();
                else if ( lhv_is_a_flag )
                    return true;
                else if ( rhv_is_a_flag )
                    return false;
                else 
                    return lhv->getName() < rhv->getName();
            }
        };


        void ShortUsage(TCLAP::CmdLineInterface& c, ::std::ostream& os) const
        {
            std::list<TCLAP::Arg*> args = c.getArgList();
            std::string name = c.getProgramName();
            TCLAP::XorHandler xorh = c.getXorHandler();
            std::vector< std::vector<TCLAP::Arg*> > xors = xorh.getXorList();

            std::string s = name + " ";

            // first the xorh
            for ( size_t i = 0; i != xors.size() ; ++i )
            {
                s += " {";
                ::std::sort(xors[i].begin(), xors[i].end(), ArgFlagLess());
                for ( TCLAP::ArgVectorIterator\
                      it = xors[i].begin(); it != xors[i].end(); it++ )
                    s += (*it)->shortID() + "|";

                s[s.length()-1] = '}';
            }

            // Сортировка стабильна.
            args.sort(ArgFlagLess());
            for ( TCLAP::ArgListIterator i = args.begin(); i != args.end();++i )
            {
                if ( !xorh.contains(*i) )
                    s += " " + (*i)->shortID();
            }

            // if the program name is too long,
            // then adjust the second line offset
            int second_line_offset = static_cast<int>(name.length()) + 2;
            if ( second_line_offset > 75/2 )
                second_line_offset = static_cast<int>(75/2);

            TCLAP::StdOutput::spacePrint(os, s, 75, 3, second_line_offset);
        }


        void LongUsage(TCLAP::CmdLineInterface& c, std::ostream& os) const
        {
            std::list<TCLAP::Arg*> args = c.getArgList();
            std::string message = c.getMessage();
            TCLAP::XorHandler xorh = c.getXorHandler();
            std::vector< std::vector<TCLAP::Arg*> > xors = xorh.getXorList();

            // first the xorh
            for ( size_t i = 0; i != xors.size(); ++i )
            {
                ::std::sort(xors[i].begin(), xors[i].end(), ArgFlagLess());
                for ( TCLAP::ArgVectorIterator it = xors[i].begin();
                      it != xors[i].end(); ++it )
                {
                    TCLAP::StdOutput::spacePrint(os, (*it)->longID(), 75, 3, 3);
                    TCLAP::StdOutput::spacePrint(
                        os, (*it)->getDescription(), 75, 5, 0);

                    if ( it+1 != xors[i].end() )
                        TCLAP::StdOutput::spacePrint(os, "-- OR --", 75, 9, 0);
                }
                os << std::endl << std::endl;
            }

            // then the rest
            args.sort(ArgFlagLess());
            for ( TCLAP::ArgListIterator \
                  it = args.begin(); it != args.end(); ++it )
            {
                if ( !xorh.contains(*it) )
                {
                    TCLAP::StdOutput::spacePrint(os, (*it)->longID(), 75, 3, 3);
                    TCLAP::StdOutput::spacePrint(
                        os, (*it)->getDescription(), 75, 5, 0);
                    os << std::endl;
                }
            }

            os << std::endl;
            TCLAP::StdOutput::spacePrint(os, message, 75, 3, 0);
        }
    };


    int CreateManager(const ::std::string& arg,
                      const ::std::string& proxy,
                      const ::std::string& include_matcher,
                      const ::std::string& exclude_matcher,
                      const ::std::string& yield_matcher,
                      const size_t urls_hash_table_size,
                      const bool grab_nofollow,
                      const bool extract_text,
                      const ::std::string& user_agent,
                      const ::std::string& referer,
                      const ::std::string& cookies,
                      const size_t connect_timeout,
                      const size_t max_page_download_time,
                      const size_t min_page_download_speed,
                      const size_t max_num_redirects)
    {
        try
        {
            // важен типизированный NULL, иначе выбирается другая перегрузка.
            const char* cp = NULL;
            const icu::UnicodeString u_i_matcher(include_matcher.data(), cp);
            const icu::UnicodeString u_e_matcher(exclude_matcher.data(), cp);
            const icu::UnicodeString u_y_matcher(yield_matcher.data(), cp);

            if ( u_i_matcher.isBogus() ||
                 u_e_matcher.isBogus() ||
                 u_y_matcher.isBogus())
            {
                ERR("Couldn't decode url matchers");
                return -1;
            }

            manager_ = new Manager(u_i_matcher,
                                   u_e_matcher,
                                   u_y_matcher,
                                   urls_hash_table_size);

            manager_->set_proxy(proxy);
            manager_->set_grab_nofollow(grab_nofollow);
            manager_->set_extract_text(extract_text);
            manager_->set_user_agent(user_agent);
            manager_->set_referer(referer);
            manager_->set_cookie(cookies);
            manager_->set_connect_timeout(connect_timeout);
            manager_->set_timeout(max_page_download_time);
            manager_->set_low_speed_limit(min_page_download_speed, 10);
            manager_->set_max_num_redirects(max_num_redirects);

            return AddUrlsToManager(arg);
        }
        catch(libv::PrecondException& e)
        {
            ERR("Manager precondition failed: " << e.what());
            return -1;
        }
        catch(UrlConverterException& e)
        {
            ERR("Input url Ascii2Unicode conversion failed: " << e.what());
            return -1;
        }
        catch(::std::exception& e)
        {
            ERR("CreateManager() caught unexpected exception: " << e.what());
            return -1;
        }
    }


    // arg или строка вида "[url] [max_depth]" или имя файла, содержащего
    // такие строки.
    int AddUrlsToManager(const ::std::string& arg)
    {
        const libv::CIString ci_arg(arg.begin(), arg.end());
        if ( ci_arg.find_first_of("http://") == 0 )
        {
            // важен типизированный NULL, иначе выбирается другая перегрузка.
            const char* cp = NULL;
            const icu::UnicodeString u_arg(arg.data(), cp);

            if ( u_arg.isBogus() || u_arg.isEmpty() )
            {
                ERR("Couldn't decode argument");
                return -1;
            }

            UnicodeUrl url;
            int32_t max_depth;
            if ( ParseUrlWithMaxDepth(u_arg, &url, &max_depth) == -1 )
                return -1;

            manager_->AddUrl(url, max_depth);
            return 0;
        }
        else
        {
            ::std::ifstream urls_file(arg.c_str(), std::ios::in);

            ::std::vector<char> bom = utf8_bom_;
            ::std::vector<char> bytes(utf8_bom_.size(), 0);
            urls_file.read(&bytes[0], utf8_bom_.size());

            if ( !::std::equal(bytes.begin(), bytes.end(), bom.begin()) )
                urls_file.seekg(0);

            size_t line_num = 1;
            while ( urls_file.good() )
            {
                ::std::string line;
                ::std::getline(urls_file, line);
                if ( line.empty() )
                    continue;

                const libv::CIString ci_line(line.begin(), line.end());
                if ( ci_line.find_first_of("http://") != 0 )
                {
                    ERR("No url found at " << arg << ":" << line_num);
                    return -1;
                }

                const UnicodeUrl u_line(line.c_str(), -1, "utf-8");

                if ( u_line.isBogus() || u_line.isEmpty() )
                {
                    ERR("Bad line " << arg << ":" << line_num << " found");
                    return -1;
                }

                UnicodeUrl url;
                int32_t max_depth;
                if ( ParseUrlWithMaxDepth(u_line, &url, &max_depth) == -1 )
                    return -1;

                manager_->AddUrl(url, max_depth);
                line_num++;
            }

            if ( !urls_file.eof() )  // чтение завершилось не из-за EOF
            {
                ERR("Error while reading " << arg);
                urls_file.close();
                return -1;
            }

            urls_file.close();
            return 0;
        }
    }


    int ParseUrlWithMaxDepth(const icu::UnicodeString& u_arg,
                             icu::UnicodeString* url,
                             int32_t* max_depth)
    {
        const int32_t o = u_arg.indexOf(' ');
        if ( o == -1 || o == u_arg.length() - 1 )
        {
            ERR("Max depth not found");
            return -1;
        }

        *max_depth = 0;
        for ( int32_t i = o + 1 ; i != u_arg.length() ; ++i )
        {
            if ( ! ( '0' <= u_arg[i] && u_arg[i] <= '9' ) )
            {
                ERR("Non-digit character found where max depth was expected");
                return -1;
            }

            *max_depth *= 10;
            *max_depth += u_arg[i] - '0';
        }

        u_arg.extract(0, o, *url);
        return 0;
    }


    int CreateDownloaders(const size_t num_downloaders,
                          const uint32_t download_pause)
    {
        if ( num_downloaders == 0 )
        {
            ERR("Num of downloader threads must be greater than zero");
            return -1;
        }

        for ( size_t i = 0 ; i != num_downloaders ; ++i )
        {
            downloaders_.push_back(
                new Manager::Downloader(manager_->GetDownloader()));
        }

        download_pause_ = download_pause;
        return 0;
    }


    int CreateExtractors(const size_t num_extractors)
    {
        if ( num_extractors == 0 )
        {
            ERR("Num of extractor threads must be greater than zero");
            return -1;
        }

        for ( size_t i = 0 ; i != num_extractors ; ++i )
        {
            extractors_.push_back(
                new Manager::Extractor(manager_->GetExtractor()));
        }
        return 0;
    }


    int ConnectToSocket(const ::std::string& server_port,
                        const long& timeout,
                        const size_t max_reconnect_attempts)
    {
        const size_t p =  server_port.find(':');
        if ( p == ::std::string::npos || p == 0 || p == server_port.size() - 1 )
        {
            ERR("Bad socket address");
            return -1;
        }

        const ::std::string server(server_port.data(), p);
        const ::std::string port(server_port.data() + p + 1);

        try
        {
            socket_sender_ = SocketSender(server, port, timeout);
            save_to_socket_ = true;
            max_reconnect_attempts_ = max_reconnect_attempts;
            return 0;
        }
        catch(SocketSenderException& e)
        {
            ERR("Couldn't connect to " << server << ":" << port
                                       << ": " << e.what());
        }
        catch(::std::exception& e)
        {
            ERR("ConnectToSocket() caught unexpected exception: " << e.what());
        }

        return -1;
    }


    /** Создание файла, куда будут построчно заноситься выгруженные адреса.
      * Закрытие файла — в деструкторе.
      * Метод непотокобезопасен и не генерирует исключения. */
    int CreateUrlsFile()
    {
        PRECOND(save_type_ == SAVE_URLS_ONLY && save_to_folder_);
        ::std::stringstream s;
        s << folder_ << '\\';
        s.width(15);
        s.fill('0');
        s << file_base_ << ".urls";

        const ::std::vector<char> bom = utf8_bom_;

        urls_file_.open(s.str().c_str(), ::std::ios::out |
                                         ::std::ios::binary |
                                         ::std::ios::trunc);
        urls_file_.write(&bom[0],
                         libv::IntegerCast< ::std::streamsize >(bom.size()));
        urls_file_.flush();

        if ( !urls_file_.good() )
        {
            ERR("Urls file creation error");
            return -1;
        }

        urls_file_.exceptions(::std::ofstream::failbit |
                              ::std::ofstream::badbit);
        return 0;
    }


    /** Создание директории по заданному пути folder.
      * Если объект с таким путем существует, метод успешно завершается
      * и не проверяет соответствует ли этот путь файлу или папке.
      * Рекомендуется задавать имена с использованием только ASCII-символов. */
    int CreateFolder(const ::std::string& folder)
    {
#ifdef _WIN32
        const int r = _mkdir(folder.c_str());
#else
        const int r = mkdir(folder.c_str(), 0777);
#endif

        if ( r == 0 )
        {
            LOG("Successfully created directory");

            save_to_folder_ = true;
            folder_ = folder;
            return 0;
        }
        else if ( r != 0 && errno == EEXIST )
        {
            LOG("Directory seems to have been created");

            save_to_folder_ = true;
            folder_ = folder;
            return 0;
        }
        else
        {
            ERR("Couldn't create directory " << folder << ": "
                << libv::StrError());
            return -1;
        }
    }

    bool IsManager() const
    {
        return omp_get_thread_num() == 0;
    }


    bool IsDownloader() const
    {
        return omp_get_thread_num() - 1 <
               libv::IntegerCast<int>(downloaders_.size());
    }

    int DownloaderNum() const
    {
        assert(IsDownloader());
        return omp_get_thread_num() - 1;
    }

    Manager::Downloader* MyDownloader() const
    {
        return downloaders_[DownloaderNum()];
    }


    bool IsExtractor() const
    {
        return omp_get_thread_num() -
                   1 - libv::IntegerCast<int>(downloaders_.size()) <
               libv::IntegerCast<int>(extractors_.size());
    }

    int ExtractorNum() const
    {
        assert(IsExtractor());
        return omp_get_thread_num() -
               1 - libv::IntegerCast<int>(downloaders_.size());
    }

    Manager::Extractor* MyExtractor() const
    {
        return extractors_[ExtractorNum()];
    }


    int ManagerCycle()
    {
        assert(IsManager());
        while ( continue_jobs_ )
        {
            try
            {
                if ( manager_->num_pages() > 0 )
                {
                    const Manager::Page p = manager_->PopPage();
                    ::std::vector<char> page_output;
                    ::std::string ext;

                    if ( save_type_ == SAVE_URLS_ONLY )
                    {
                        page_output = UnicodeString2UTF8(p.url);
                    }
                    else if ( save_type_ == SAVE_AS_PLAIN_TEXT )                       
                    {
                        page_output = MakePlainText(p);
                        ext = "txt";
                    }
                    else if ( save_type_ == SAVE_ORIGINAL_FILES )
                    {
                        page_output = p.data;
                        ext = MimeType2Extension(p.mime_type);
                    }
                    else if ( save_type_ == SAVE_AS_XML )
                    {
                        page_output = MakeXml(p);
                        ext = "xml";
                    }
                    else 
                        assert(0);

                    if ( save_to_folder_ )
                    {
                        if ( save_type_ == SAVE_URLS_ONLY )
                        {
                            LOG("Adding url to urls file");
                            AppendToUrlsFile(page_output);
                            MSG("Url has been successfully saved");
                        }
                        else
                        {
                            LOG("Saving page into file");
                            Save(page_output, ext);
                            MSG("Plain has been successfully saved");
                        }
                    }

                    if ( save_to_socket_ )
                    {
                        LOG("Sending page to server");
                        Send(page_output);
                        MSG("Page has been successfully sent");
                    }

                    continue;
                }


                // Готовых страниц нет. //

                if ( save_to_socket_ )
                {
                    LOG("There is no pages, sending keep-alive");
                    SendKeepAlive();
                    MSG("Keep-alive successfully sent");
                }

                if ( manager_->IsAllDone() )
                {
                    continue_jobs_ = 0;
                    return 0;
                }

                // Если страниц нет, не нужно постоянно обращаться к менеджеру.
                libv::MilliSleep(1000);
            }
            catch(SocketSenderException& e)
            {
                ERR("Socket send error: " << e.what());

                continue_jobs_ = 0;  // присваивание атомарно.
                return -1;
            }
            catch(::std::ofstream::failure& e)
            {
                ERR("File saving error: " << e.what());

                continue_jobs_ = 0;
                return -1;
            }
            catch(::std::exception& e)
            {
                ERR("Manager thread caught unexpected exception: " << e.what());

                continue_jobs_ = 0;
                return -1;
            }
        }

        // Выход по continue_jobs_ == 0, установленному не менеджером.
        return -1;
    }


    void DownloaderCycle()
    {
        assert(IsDownloader());
        try
        {
            while ( continue_jobs_ )
            {
                if ( MyDownloader()->TryDoOneJob() == false )
                    MSG("Downloader"<< DownloaderNum() << ": there is no jobs");

                libv::MilliSleep(download_pause_ * 1000);
            }
        }
        catch(::std::exception& e)
        {
            // Программа спроектирована так, что метод
            // MyDownloader()->TryDoOneJob() не должен генерировать исключения.
            // Но если в случае ошибки это все же произойдет, то ловим его
            // здесь, так как запрещено выбрасывать исключение из openmp
            // и завершаем программу.
            ERR("DownloaderCycle() caught unexpected exception: " << e.what());
            continue_jobs_ = 0;  // присваивание атомарно.
            return;
        }
    }


    void ExtractorCycle()
    {
        assert(IsExtractor());
        try
        {
            while ( continue_jobs_ )
            {
                if ( MyExtractor()->TryDoOneJob() == false )
                {
                    MSG("Extractor" << ExtractorNum() << ": there is no jobs");
                    libv::MilliSleep(1000);
                }
            }
        }
        catch(::std::exception& e)
        {
            // Программа спроектирована так, что метод
            // MyExtractor()->TryDoOneJob() не должен генерировать исключения.
            // Но если в случае ошибки это все же произойдет, то ловим его
            // здесь, так как запрещено выбрасывать исключение из openmp
            // и завершаем программу.
            ERR("ExtractorCycle() caught unexpected exception: " << e.what());
            continue_jobs_ = 0;  // присваивание атомарно.
            return;
        }
    }


    ::std::string MimeType2Extension(const MimeType& mime_type) const
    {
        if ( mime2ext_.count(mime_type) == 0 )
            return "file";
        return mime2ext_.find(mime_type)->second;
    }


    ::std::vector<char> UnicodeString2UTF8(const icu::UnicodeString& s) const
    {
        if ( s.isEmpty() ) return ::std::vector<char>();

        // Сначала первая неудачная попытка, что бы узнать конечный размер.
        ::std::vector<char> v_str(1);
        const int32_t exp_len = s.extract(0, s.length(), &v_str[0], 1, "utf-8");

        v_str.reserve(exp_len);
        v_str.resize(exp_len);

        const int32_t act_len =
            s.extract(0, s.length(), &v_str[0], UI32(v_str.size()), "utf-8");
        VERIFY(act_len == I32(v_str.size()));

        return v_str;
    }


    /** Создание простого текста из переданной страницы.
      * Формат (используется кодировка UTF-8):
      * <BOM>GUID\n
      * URL\n
      * PARENT_URL\n
      * \n
      * TEXT.
      * Где <BOM> — byte order mark кодировки UTF-8 (3 байта)
      * GUID — идентификатор страницы,
      * URL — адрес страницы,
      * PARENT_URL — адрес страницы, ссылавшейся на данную,
      * TEXT — текст страницы.
      * Метод не потокобезопасен и не генерирует исключения. */
    ::std::vector<char> MakePlainText(const Manager::Page& page) const
    {
        ::std::vector<char> result;

        // Для большинства страниц каждый символ unicode, кодируемый в utf-8
        // будет занимать не более 2х байт и перевыделения памяти не будет.
        result.reserve( ( utf8_bom_.size() + 
                          page.url.length() +
                          page.url.length() +
                          page.parent_url.length() +
                          page.text.length() ) * 2 );

        result = utf8_bom_;

        const ::std::vector<char> u8_url = UnicodeString2UTF8(page.url);
        result.insert(result.end(), u8_url.begin(), u8_url.end());
        result.push_back('\n');

        // На данный момент GUID совпадает с URL.
        result.insert(result.end(), u8_url.begin(), u8_url.end());
        result.push_back('\n');

        const ::std::vector<char> u8_purl = UnicodeString2UTF8(page.parent_url);
        result.insert(result.end(), u8_purl.begin(), u8_purl.end());
        result.push_back('\n');
        result.push_back('\n');

        const ::std::vector<char> u8_text = UnicodeString2UTF8(page.text);
        result.insert(result.end(), u8_text.begin(), u8_text.end());

        return result;
    }


    /** Создание XML-документа из переданной страницы.
      * Ниже приводится формат, все значения кодируются Base64.
      * @code
      * <?xml version="1.0" encoding="utf-8" ?>
      * <document title="scrw document">
      *     <attributes>
      *         <guid><![CDATA[…]]></guid>
      *         <timestamp>2010-12-16 22:37:38</timestamp>
      *         <group name="other">
      *             <attribute name="url"><![CDATA[…]]></attribute>
      *             <attribute name="parent"><![CDATA[…]]>attribute>
      *         </group>
      *     </attributes>
      *     <blocks>
      *         <block type="text"><![CDATA[…]]></block>
      *     </blocks>
      * </document>
      * @endcode */
    ::std::vector<char> MakeXml(const Manager::Page& page) const
    {
        ::std::string result;

        result +=
            "<?xml version=\"1.0\" encoding=\"utf-8\" ?>\n"
            "<document title=\"scrw document\">\n"
            "    <attributes>\n"
            "        <guid><![CDATA[";

        result += base64_.Encode(UnicodeString2UTF8(page.url));

        result +=
            "]]></guid>\n"
            "        <timestamp>";

		result += FormattedTime("%Y-%m-%d %H:%M:%S");

        result +=
            "</timestamp>\n"
            "        <group name=\"other\">\n"
            "            <attribute name=\"url\"><![CDATA[";

        result += base64_.Encode(UnicodeString2UTF8(page.url));

        result +=
            "]]></attribute>\n"
            "            <attribute name=\"parent\"><![CDATA[";

        result += base64_.Encode(UnicodeString2UTF8(page.parent_url));

        result +=
            "]]></attribute>\n"
            "        </group>\n"
            "    </attributes>\n"
            "    <blocks>\n"
            "        <block type=\"text\"><![CDATA[";

        result += base64_.Encode(UnicodeString2UTF8(page.text));

        result +=
            "]]></block>\n"
            "    </blocks>\n"
            "</document>\n";

        return ::std::vector<char>(result.begin(), result.end());
    }


    /** Передача keep-alive сообщения по сокету. */
    void SendKeepAlive()
    {
        Send(::std::vector<char>());
    }


    /** Передача строки по сокету.
      * Формат передачи: N<пробел>DATA,
      * где N — текстовое представление размера данных DATA,
      * DATA — передаваемая строка.
      * В случае разрыва соединения производится переподключение,
      * но не более max_reconnect_attempts_ раз, после чего метод
      * генерирует исключение ConnectionLostException.
      * @exception SocketSenderException.
      * Метод не потокобезопасен и генерирует исключения. */
    void Send(const ::std::vector<char>& data)
    {
        ::std::stringstream header;
        header << data.size() << " ";

        bool successfully_sent = false;
        size_t reconnect_attempts = 0;
        while ( successfully_sent == false )
        {
            try
            {
                socket_sender_.Send(header.str().c_str(), header.str().size());

                if ( !data.empty() )  // если пустое — это KeepAlive
                    socket_sender_.Send(&data[0], data.size());

                successfully_sent = true;
            }
            catch(ConnectionLostException&)
            {
                if ( ++reconnect_attempts > max_reconnect_attempts_ )
                {
                    ERR("Too many reconnect attempts");
                    throw;
                }

                // Пробуем переподсоедениться, если не получится,
                // сгенерированное исключение пойдет дальше.
                socket_sender_.Reconnect();
                LOG("Connection lost, reconnecting");
            }
        }
    }


    /** Сохранение данных data в папку, заданную при запуске в новый файл с
      * инкрементированным порядковым номером.
      * Метод не потокобезопасен и генерирует исключения.
      * @exception ::std::ofstream::failure. */
    void Save(const ::std::vector<char>& data, const ::std::string& ext)
    {
        ::std::stringstream s;
        s << folder_ << '\\';
        s.width(15);
        s.fill('0');
        s << file_base_;

        s << '.' << (file_index_++).Value() << '.' << ext;

        // Send генерирует исключения, Save следует не нарушать архитектуру.
        ::std::ofstream f(s.str().c_str(), ::std::ios::binary|::std::ios::out);
        f.exceptions(::std::ofstream::failbit | ::std::ofstream::badbit);
        f.write(&data[0], libv::IntegerCast< ::std::streamsize >(data.size()));
        f.flush();
        f.close();
    }


    /** Добавление url в файл с адресами. Метод используется в случае, если
      * используется режим выгрузки только url (save_urls_only);
      * Метод не потокобезопасен и генерирует исключения.
      * @exception ::std::ofstream::failure. */
    void AppendToUrlsFile(const ::std::vector<char>& data)
    {
        urls_file_.write(&data[0],
                         libv::IntegerCast< ::std::streamsize >(data.size()));
        urls_file_ << ::std::endl; // это еще и flush
    }



    Manager* manager_;
    ::std::vector<Manager::Downloader*> downloaders_;
    ::std::vector<Manager::Extractor*> extractors_;

    int save_type_;

    ::std::vector<char> utf8_bom_;
    ::std::map<MimeType, ::std::string> mime2ext_;

    ::std::ofstream urls_file_;
    Base64 base64_;

    bool save_to_socket_;
    SocketSender socket_sender_;
    size_t max_reconnect_attempts_;
    uint32_t download_pause_;

    bool save_to_folder_;
    ::std::string folder_;
    uint64_t file_base_;

    libv::Atomic32 file_index_;

    // Явно 32-битная переменная, так как чтение и запись — атомарны.
    mutable libv::AtomicType32 continue_jobs_;
};



int Main(int argc, char** argv)
{
    Program program;
    const int r = program.ParseArguments(argc, argv);
    if ( r != 0 )
        return r;

    const int p = program.Do();
    if ( p != 0 )
        ::std::cout << "Something went wrong, please read log" << ::std::endl;
    else
        ::std::cout << "Crawling has been finished" << ::std::endl;
    return p;
}

}  // scrw

#endif

int main(int argc, char **argv)
{
    return scrw::Main(argc, argv);
}
